package com.cnzhy.topic;

import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * ActiveMQ中的消费者(Consumer)
 *
 * @author dengp
 */
public class TopicConsumer {

    public void reciveHelloFormActiveMq() {
        TopicSession session = null;
        TopicConnection conn = null;
        try {
            TopicConnectionFactory factory = new ActiveMQConnectionFactory("admin", "admin", "tcp://www.zhaoyan5.com:61616");
            conn = factory.createTopicConnection();
            conn.start();
            session = conn.createTopicSession(true, Session.AUTO_ACKNOWLEDGE);
            // 创建消息队列
            Topic topic = session.createTopic("test-topic");
            // 创建消息接受者
            MessageConsumer consumer = session.createConsumer(topic);
            consumer.setMessageListener(msg -> {
                if (msg != null) {
                    MapMessage map = (MapMessage) msg;
                    try {
                        System.out.println(map.getString("name") + "接收#" + map.getString("address"));
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 休眠100s再关闭
            Thread.sleep(1000 * 100);
            // 提交会话
            session.commit();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println("访问ActiveMQ服务发生错误!!");
        } finally {
            try {
                // 回收会话资源
                if (null != session)
                    session.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
            try {
                // 回收链接资源
                if (null != conn)
                    conn.close();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }
}
